package com.bw.app.dwm;

import com.alibaba.fastjson.JSONObject;
import com.bw.been.OrderWide;
import com.bw.func.AsyncDatabaseRequest;
import com.bw.func.AsyncDimFunction;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.ParseException;
import java.util.concurrent.TimeUnit;

public class AllWide {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String groupId="AllWide";
        String topic="dws_info_wide";

        FlinkKafkaConsumer<String> source = MyKafkaUtil.getKafkaSource(groupId, topic);
        DataStreamSource<String> info_wide = env.addSource(source);
//        info_wide.print();
        SingleOutputStreamOperator<OrderWide> InfoWide = info_wide.map(data -> JSONObject.parseObject(data, OrderWide.class));
//        InfoWide.print();
        //关联base_house
        SingleOutputStreamOperator<OrderWide> HouseWide = AsyncDataStream.unorderedWait(InfoWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_house") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getIshouse();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setHouser_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        HouseWide.print("HouseWide>>>>>>>>>>>>>>>>");
        //关联base_marry
        SingleOutputStreamOperator<OrderWide> MarryWide = AsyncDataStream.unorderedWait(HouseWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_marry") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getMarryid();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setMarry_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        MarryWide.print("MarryWide>>>>>>>>>>>>>>>>");
        //关联base_start
        SingleOutputStreamOperator<OrderWide> StartWide = AsyncDataStream.unorderedWait(MarryWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_start") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getIsstart();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setStart_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        StartWide.print("StartWide>>>>>>>>>>>>>>>>");
        //关联base_vip
        SingleOutputStreamOperator<OrderWide> VipWide = AsyncDataStream.unorderedWait(StartWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_vip") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getIsmoney();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setVip_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        VipWide.print("VipWide>>>>>>>>>>>>>>>>");
        //关联base_xueli
        SingleOutputStreamOperator<OrderWide> XueliWide = AsyncDataStream.unorderedWait(VipWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_xueli") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getXueli();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setXueli_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        XueliWide.print("XueliWide>>>>>>>>>>>>>>>>");
        //关联base_xydj
        SingleOutputStreamOperator<OrderWide> XydjWide = AsyncDataStream.unorderedWait(XueliWide,
                new AsyncDimFunction<OrderWide>("GMALL2022_REALTIME:dim_base_xydj") {

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getXytz();
                    }

                    @Override
                    public void join(OrderWide orderWide, String dimJSON) throws ParseException {

                        if (dimJSON != null) {
                            orderWide.setXydj_tagname(dimJSON);
                        }
                    }
                },
                100,
                TimeUnit.SECONDS);
//        XydjWide.print("XydjWide>>>>>>>>>>>>>>>>");

        SingleOutputStreamOperator<String> AllWide = XydjWide.map(new MapFunction<OrderWide, String>() {
            @Override
            public String map(OrderWide orderWide) throws Exception {
                return JSONObject.toJSONString(orderWide);
            }
        });

        String dwm_topic="dwm_info_wide";
        AllWide.addSink(MyKafkaUtil.getKafkaSink(dwm_topic));

        env.execute();
    }
}